1.1.FastAPI 快速上手

@app.get("/double")
async def double(num: int):
  return {"result": num * 2}

uvicorn.run("app.server:app", host="0.0.0.0", port=port)

@app.get("/test")
async def test():
  print(f"Process {os.getpid()} handling /test")
  return {"message": "Hello World"}


@app.get("/sync_delay")
async def sync_delay(delay: int = 1):
  """同步延迟delay秒"""
  print(f"Process {os.getpid()} handling /sync_delay")
  time.sleep(delay)
  return {"hello": "world"}


@app.get("/async_delay")
async def async_delay(delay: int = 1):
  """异步延迟delay秒"""
  print(f"Process {os.getpid()} handling /async_delay")
  await asyncio.sleep(delay)
  return {"hello": "world"}

import Axios from "axios";
import {Button, Space} from "antd";

export default function () {

  /*调用test接口，打印消耗的时间*/
  async function request_test() {
    const startTime = Date.now();
    const resp = await Axios.get('http://127.0.0.1:7002/test');
    console.log('request_test result:', resp.data);
    console.log(`耗时：${((Date.now() - startTime) / 1000).toFixed(2)}s`);
  }

  /*调用doubao链*/
  async function request_doubao_endpoint() {
    const resp = await Axios.post(
      'http://127.0.0.1:7002/doubao/invoke',
      {
        input: { messages: [{ role: 'user', content: '写一个关于海洋的小作文，300字' }], },
        config: {},
        kwargs: {},
      }
    );
    console.log('request_doubao_endpoint result:', resp.data);
  }

  /*调用同步阻塞接口*/
  async function syncDelay(delay = 3) {
    const resp = await Axios.get(`http://127.0.0.1:7002/sync_delay?delay=${delay}`);
    console.log('syncDelay result:', resp.data);
  }

  /*调用异步阻塞接口*/
  async function asyncDelay() {
    const resp = await Axios.get('http://127.0.0.1:7002/async_delay?delay=3');
    console.log('asyncDelay result:', resp.data);
  }

  /*同时调用“同步阻塞接口”以及“test”接口*/
  async function syncDelayAndTest() {
    syncDelay();
    request_test();
  }

  /*同时调用“异步阻塞接口”以及“test”接口*/
  async function asyncDelayAndTest() {
    asyncDelay();
    request_test();
  }

  /*同时调用“豆包”接口以及test接口*/
  async function doubaoAndTest() {
    request_doubao_endpoint();
    request_test();
  }

  return (
    <div style={{ padding: '1em' }}>
      <Space>
        <Button onClick={request_test}>调用test接口</Button>
        <Button onClick={syncDelayAndTest}>请求接口：同步阻塞、test</Button>
        <Button onClick={asyncDelayAndTest}>请求接口：异步阻塞、test</Button>
        <Button onClick={doubaoAndTest}>请求接口：doubao、test</Button>
      </Space>
    </div>
  );
}

uvicorn.run("app.server:app", host="0.0.0.0", port=port, workers=2)

/*同时调用“同步阻塞接口”以及“test”接口*/
async function syncDelayAndTest() {
  syncDelay(2);
  syncDelay(3);
  request_test();
}

Process 19884 handling /sync_delay
Process 18740 handling /test
INFO:     127.0.0.1:9869 - "GET /test HTTP/1.1" 200 OK
INFO:     127.0.0.1:9868 - "GET /sync_delay?delay=3 HTTP/1.1" 200 OK
Process 19884 handling /sync_delay
INFO:     127.0.0.1:9868 - "GET /sync_delay?delay=3 HTTP/1.1" 200 OK

Process 19884 handling /sync_delay
Process 18740 handling /sync_delay
INFO:     127.0.0.1:10066 - "GET /sync_delay?delay=2 HTTP/1.1" 200 OK
INFO:     127.0.0.1:10067 - "GET /sync_delay?delay=3 HTTP/1.1" 200 OK
Process 18740 handling /test
INFO:     127.0.0.1:10068 - "GET /test HTTP/1.1" 200 OK

http {
    upstream backend {
        server backend1.example.com;
        server backend2.example.com;
        server backend3.example.com;
    }
    server {
        listen 80;
        location / {
            proxy_pass http://backend;
        }
    }
}

@app.get("/my_stream")
async def async_delay(start: int, end: int):
  async def generate_numbers():
    current = start
    while current <= end:
      yield json.dumps({"number": current}) + '\n'
      await asyncio.sleep(0.2)
      current += 1

  return StreamingResponse(generate_numbers(), media_type="application/x-ndjson")

import json
import time
from typing import List

from fastapi.responses import StreamingResponse
from fastapi import FastAPI, APIRouter
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable


def format_ai_message(ai_message: AIMessage):
  return {
    "choices": [{
      "finish_reason": "stop",
      "index": 0,
      "message": {
        "content": ai_message.content,
        "role": "assistant"
      }
    }],
    "created": int(time.time()),
    "id": ai_message.id,
    "usage": ai_message.response_metadata.get('token_usage')
  }


def add_async_routes(app: FastAPI, path: str, runnable: Runnable):
  _router = APIRouter(prefix=path, tags=[path])

  @_router.post("ainvoke")
  async def ainvoke(input: runnable.input_schema):
    ai_message: AIMessage = await runnable.ainvoke(input.model_dump())
    return format_ai_message(ai_message)

  @_router.post("astream")
  async def astream(input: runnable.input_schema):  # 这里 input_schema 需确保已正确定义
    async def generator_function():
      # 初始化一个结构，用于构建最终返回的包含 choices 等的数据
      result_template = {
        "choices": [{"delta": {}, "index": 0}],
        "created": time.time(),
        "id": "",
        "usage": None
      }
      async for chunk in runnable.astream(input.model_dump()):
        # 更新 delta 里的内容
        result_template["choices"][0]["delta"]["content"] = chunk.content
        result_template["choices"][0]["delta"]["role"] = 'assistant'

        if chunk.response_metadata.get('finish_reason') is not None:
          result_template["choices"][0]["delta"]['finish_reason'] = chunk.response_metadata.get('finish_reason')

        result_template['id'] = chunk.id
        result_template['created'] = int(time.time())
        yield f"data: {json.dumps(result_template, ensure_ascii=False)}\n\n"
      yield "data: [DONE]\n\n"

    return StreamingResponse(generator_function(), media_type="text/event-stream")

  @_router.post("abatch")
  async def abatch(inputs: list[runnable.input_schema]):
    ai_message_list: List[AIMessage] = await runnable.abatch([input.model_dump() for input in inputs])
    return [format_ai_message(ai_message) for ai_message in ai_message_list]

  app.include_router(_router)

add_async_routes(
  app=app,
  runnable=RunnableLambda(lambda x: x['messages']) | create_llm().with_types(input_type=ModelInputSchema),
  path="/async_doubao")

add_async_routes(
  app=app,
  runnable=RunnableLambda(lambda x: x['messages']) | create_llm("huoshan-deepseek-r1").with_types(input_type=ModelInputSchema),
  path="/async_doubao_deepseek_r1")